package buaa.qel.rpc.common.serializer.kryo;

import buaa.qel.rpc.common.serializer.Serializer;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.objenesis.strategy.StdInstantiatorStrategy;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * kryo实现序列化/反序列化工具
 */
public class KryoSerializer extends Serializer {
    /**
     * kryo不是线程安全的，会对序列化和反序列化的状态进行保存，所以多个线程公用一个kryo对象会造成
     * 线程安全问题。因此使用threadlocal给每条线程绑定一个kryo对象
     */
    private final ThreadLocal<Kryo> kryoThreadLocal = new ThreadLocal<Kryo>(){
        @Override
        protected Kryo initialValue() {
            Kryo kryo = new Kryo();
            /**
             * Registration即注册，kryo在序列化对象时，首先会序列化其类的全限定名，由于我们通常序列化
             * 的对象都是有限范围内的类的实例，这样重复序列化同样的类的全限定名是低效的。通过注册kryo可以
             * 将类的全限定名抽象为一个数字，即用一个数字代表全限定名，这样就要高效一些。
             * kryo.register(SomeClass.class);注册方法的完整签名为
             * public Registration register (Class type, Serializer serializer, int id)，
             * 我们通常只需要使用其重载方法即可public Registration register (Class type)，
             * serializer和id在kryo内部会指定。kryo.register(classToSerialize);
             * 分布式环境下可能因为序列化和反序列化同一个类时id不一致报错，所以关闭注册，牺牲一部分性能
             * kryo.register(classToSerialize);
             */
            kryo.setRegistrationRequired(false);
            /**
             * References即引用，对A对象序列化时，默认情况下kryo会在每个成员对象第一次序列化时
             * 写入一个数字，该数字逻辑上就代表了对该成员对象的引用，如果后续有引用指向该成员对象，
             * 则直接序列化之前存入的数字即可，而不需要再次序列化对象本身。
             * 这种默认策略对于成员存在互相引用的情况较有利，否则就会造成空间浪费
             * （因为每序列化一个成员对象，都多序列化一个数字），默认值为true，关闭后可能造成内存溢出
             */
            kryo.setReferences(true);
            /**
             * 设置初始化策略，如果没有默认无参构造器，那么就需要设置此项,使用此策略构造一个无参构造器
             */
            kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
            return kryo;
        }
    };

    public Kryo getKryo(){
        return this.kryoThreadLocal.get();
    }

    @Override
    public byte[] serialize(Object object) {
        Kryo kryo = getKryo();
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);
        kryo.writeClassAndObject(output, object);
        output.close();
        return byteArrayOutputStream.toByteArray();
    }

    @Override
    public <T> Object deserialize(byte[] bytes,Class<T> clazz) {
        Kryo kryo = getKryo();
        Input input = new Input(new ByteArrayInputStream(bytes));
        return (T) kryo.readClassAndObject(input);
    }
}
